/*
 * UMediaReaderSource.cpp
 *
 *  Created on: 2020年4月25日
 *      Author: chuanjiang.zh
 */

#include "UMediaReaderSource.h"
#include "FfmpegMediaReader.h"
#include "TimeHelper.h"
#include "CLog.h"
#include "UMediaEvent.h"
#include "LoopMediaReader.h"
#include <assert.h>

static const int MAX_SYNC_TIME = 10 * 1000 * 1000;
static const AVRational PREREAD_SCALE = av_make_q(5, 2);

static const int MIN_DELAY_TIME = 5;	/// ms

static const float MAX_SCALE = 8.0;
static const float MIN_SCALE = 1.0 / 8;
static const float STEP_SCALE = 1.0 / 4;


int UMediaReaderSource::STACK_SIZE = 1024 * 1024 ;
bool UMediaReaderSource::DEBUG_ENABLED = false;




UMediaReaderSource::UMediaReaderSource():
	m_isLive(true),
	m_seekable(),
	m_refStreamIndex(),
	m_curTime(),
	m_scale(av_make_q(1, 1)),
	m_endOfStream(),
	m_loopCount(),
	m_retryDelay(1000),
	m_cutoutTime(),
	m_audioTsCustomed(),
	m_connected()
{
	comn::Thread::setStackSize(STACK_SIZE);
}

UMediaReaderSource::~UMediaReaderSource()
{
}


void UMediaReaderSource::setLoopCount(int loopCount)
{
    comn::AutoCritSec lock(m_csSource);
	m_loopCount = loopCount;
}

bool UMediaReaderSource::isLoopEnabled()
{
    comn::AutoCritSec lock(m_csSource);
    return m_loopCount < 0 || m_loopCount > 1;
}


int UMediaReaderSource::open(const char* url, const char* params)
{
	if (!url || strlen(url) == 0)
	{
		return EINVAL;
	}

	if (isOpen())
	{
		close();
	}

	m_url = url;
	m_params = params;

	m_endOfStream = false;

    if (isRunning())
    {
        CLog::warning("UMediaReaderSource::open. already open\n");
    }
    else
    {
        comn::CommandThread::start();
    }

	postCommand(kOpen);

	return 0;
}

void UMediaReaderSource::close()
{
	int state = getState();
	if ((state != UMEDIA_STATE_STOPPED) && (state != UMEDIA_STATE_CLOSED))
	{
		stopStream();
	}

	if (isOpen())
	{
		postCommand(kClose);
	}

	comn::CommandThread::stop();

	m_url.clear();

	m_endOfStream = false;
	m_isLive = true;
	m_seekable = false;

    m_curTime = 0;
}

bool UMediaReaderSource::isOpen()
{
	return isRunning();
}

bool UMediaReaderSource::getMediaFormat(UMediaFormat& fmt)
{
	if (!m_reader)
	{
		return false;
	}

    return BaseMediaSource::getMediaFormat(fmt);
}

int UMediaReaderSource::startStream()
{
	postCommand(kPlay);
	return 0;
}

int UMediaReaderSource::pauseStream()
{
	postCommand(kPause);
	return 0;
}

void UMediaReaderSource::stopStream()
{
	postCommand(kStop);
}

bool UMediaReaderSource::seekable()
{
    comn::AutoCritSec lock(m_csSource);
	return m_seekable;
}

bool UMediaReaderSource::seek(int64_t pos)
{
	if (pos < 0)
	{
		return false;
	}

	{
		comn::AutoCritSec lock(m_csSource);
		m_curTime = pos;
	}

	postCommand(kSeek, pos);
	return true;
}

int64_t UMediaReaderSource::getTime()
{
    comn::AutoCritSec lock(m_csSource);
	return m_curTime;
}

bool UMediaReaderSource::setScale(float scale)
{
	if ((scale < MIN_SCALE) || (scale > MAX_SCALE))
	{
		return false;
	}

	AVRational r = av_d2q(scale, 32);

	postCommand(kSetScale, r.num, r.den);

	return true;
}

float UMediaReaderSource::getScale()
{
    comn::AutoCritSec lock(m_csSource);
	return (float)av_q2d(m_scale);
}

void UMediaReaderSource::onCommand(Command& cmd)
{
	if (cmd.type == kOpen)
	{
		handleOpen();
	}
	else if (cmd.type == kClose)
	{
		handleClose();
	}
	else if (cmd.type == kPlay)
	{
		handlePlay();
	}
	else if (cmd.type == kPause)
	{
		handlePause();
	}
	else if (cmd.type == kStop)
	{
		handleStop();
	}
	else if (cmd.type == kSeek)
	{
		int64_t offset = cmd.at(0).toLong();
		handleSeek(offset);
	}
	else if (cmd.type == kSetScale)
	{
		AVRational scale;
		scale.num = cmd.at(0).toInt();
		scale.den = cmd.at(1).toInt();

		handleSetScale(scale);

		emitMediaEvent(UMEDIA_EVENT_SCALE);
	}
	else if (cmd.type == kStep)
	{
		handleStep();
	}
	else if (cmd.type == kNextKeyFrame)
	{
		handleNextKeyFrame();
	}
	else if (cmd.type == kFlush)
	{
		handleFlush();
	}
	else if (cmd.type == kSetProp)
	{
		std::string key = cmd.at(0).toString();
		std::string value = cmd.at(1).toString();
		handleSetProp(key, value);
	}
	else
	{
	}

}

void UMediaReaderSource::onIdle()
{
	if (!m_reader)
	{
		timedwait(-1);
		return;
	}

	if (m_state != UMEDIA_STATE_PLAYING)
	{
		timedwait(-1);
		return;
	}

	UMediaPacket packet;

	int rc = 0;
	rc = read(packet);

	if (rc == 0)
	{
		//assert(packet.dts() != AV_NOPTS_VALUE);
		if (!m_connected)
		{
			m_connected = true;
			emitMediaEvent(UMEDIA_EVENT_OPENED);
		}

		if (m_isLive)
		{
            m_cutoutTime = 0;
		}
        else
        {
            int ms = computeDelay(packet);
            delay(ms);
        }

		dispatchPacket(packet);
	}
	else if (rc == AVERROR_EOF)
	{
        emitMediaEvent(UMEDIA_EVENT_END);

		handlePause();
	}
	else if (rc == EAGAIN || (rc == AVERROR(EAGAIN)))
	{
        int delay = 50;
        if (m_isLive || isLoopEnabled())
        {
            int64_t clk = getClock();
            if (m_cutoutTime <= 0)
            {
                m_cutoutTime = clk;
            }
            else
            {
                int64_t duration = clk - m_cutoutTime;
                if (duration > (AV_TIME_BASE * 5))
                {
                    CLog::warning("UMediaReaderSource cutout.\n");

                    emitMediaEvent(UMEDIA_EVENT_END);

					handlePause();
                }
            }
        }

		// continue.
        if (delay > 0)
        {
            timedwait(delay);
        }
	}
	else if (rc == EINTR || rc == AVERROR(EINTR))
	{
		m_connected = false;
		emitMediaEvent(UMEDIA_EVENT_INTR);
	}
	else
	{
		char errbuf[AV_ERROR_MAX_STRING_SIZE] = { 0 };
		av_make_error_string(errbuf, AV_ERROR_MAX_STRING_SIZE, rc);

		CLog::info("UMediaReaderSource read error! code:%08x, err: %s, url:%s\n",
			rc, errbuf,
			m_url.c_str());

		emitMediaEvent(UMEDIA_EVENT_END);

        if (m_isLive || isLoopEnabled())
        {
            timedwait(0);
			delayWait(m_retryDelay);
            reopen();
        }
        else
        {
            timedwait(-1);
        }
	}
}

bool UMediaReaderSource::startup()
{
    //CLog::info("-------------------------- thread id: %d\n", comn::Thread::getCurrentThreadId());

	return true;
}

void UMediaReaderSource::cleanup()
{
	while (fetchAndHandle())
	{
		// pass
	}
}

void UMediaReaderSource::doStop()
{
	comn::CommandThread::doStop();

	m_eventDelay.post();

	if (m_reader)
	{
		m_reader->interrupt();
	}
}


void UMediaReaderSource::handleOpen()
{
    if (m_canExit)
    {
        return;
    }

    m_reader = createReader(m_url, isLoopEnabled());

	if (!m_reader)
	{
		CLog::error("source is null.\n");
		return;
	}

	emitMediaEvent(UMEDIA_EVENT_OPENING);

	int rc = m_reader->open(m_url.c_str(), m_params.c_str());
	if (rc != 0)
	{
		m_connected = false;
		emitMediaEvent(UMEDIA_EVENT_OPEN_FAILED);
		m_reader.reset();

        if ((!m_canExit) && getState() != UMEDIA_STATE_CLOSED)
        {
            timedwait(0);
			delayWait(m_retryDelay);

            postCommand(kOpen);
        }

		return;
	}

	m_connected = true;
	emitMediaEvent(UMEDIA_EVENT_OPENED);

	m_reader->getFormat(m_format);
	m_seekable = m_reader->seekable();

	m_isLive = m_reader->isLive();
	std::string fmtName = m_reader->getFormatName();
	if (fmtName == "hls")
	{
		m_isLive = false;
	}

	m_refStreamIndex = m_format.hasVideo() ? UMEDIA_TYPE_VIDEO : UMEDIA_TYPE_AUDIO;

	if (m_format.getVideoPropSize() > 0)
	{
		//
	}

	emitMediaEvent(UMEDIA_EVENT_FORMAT_READY);

	fireMediaFormat(m_format);


	if (UMEDIA_STATE_PLAYING == getState())
	{
		m_reader->play();

		resetTimePoint();

		emitMediaEvent(UMEDIA_EVENT_PLAYING);
	}
}

void UMediaReaderSource::handleClose()
{
	doCloseFile();

    setState(UMEDIA_STATE_CLOSED);
}

void UMediaReaderSource::handlePlay()
{
	if (!m_reader)
	{
		setState(UMEDIA_STATE_PLAYING);
		return;
	}

    if (UMEDIA_STATE_PLAYING == getState())
    {
        return;
    }

	m_reader->play();

	setState(UMEDIA_STATE_PLAYING);

	resetTimePoint();

	emitMediaEvent(UMEDIA_EVENT_PLAYING);
}

void UMediaReaderSource::handlePause()
{
	if (!m_reader)
	{
		return;
	}

	setState(UMEDIA_STATE_PAUSED);
	resetTimePoint();

	m_reader->pause();

	emitMediaEvent(UMEDIA_EVENT_PAUSED);
}

void UMediaReaderSource::handleStop()
{
	setState(UMEDIA_STATE_STOPPED);

	if (m_reader)
	{
		m_reader->stop();
	}

	emitMediaEvent(UMEDIA_EVENT_STOPPED);

	m_scale = av_make_q(1, 1);
	m_curTime = 0;

	resetTimePoint();
}

void UMediaReaderSource::handleSeek(int64_t offset)
{
	emitMediaEvent(UMEDIA_EVENT_SEEKING);

	int rc = m_reader->seek(offset);
	if (rc != 0)
	{
		CLog::warning("failed to seek. offset:%lld", offset);
	}

	resetTimePoint();

	emitMediaEvent(UMEDIA_EVENT_SEEKED);

	handleFlush();

}

void UMediaReaderSource::handleSetScale(AVRational scale)
{
	resetTimePoint();

	m_scale = scale;
}

void UMediaReaderSource::handleStep()
{
	if (m_state == UMEDIA_STATE_PLAYING)
	{
		handlePause();
	}

	readUntilVideoFrame();
}

void UMediaReaderSource::handleNextKeyFrame()
{
	if (m_state == UMEDIA_STATE_PLAYING)
	{
		handlePause();
	}

	readUntilVideoKeyFrame();
}


void UMediaReaderSource::handleFlush()
{
	//
}


void UMediaReaderSource::handleSetProp(std::string& key, std::string& value)
{
	//	std::string sec;
	//	std::string subName;
	//
	//	comn::StringUtil::split(key, comn::IniProperties::SEPERATOR, sec, subName);
}



void UMediaReaderSource::emitMediaEvent(int event)
{
	//if (type != UMEDIA_EVENT_PACKET)
	//CLog::info("media event. type:%d. name:%s\n", type, Utility::getEventName(type));

	if (event == UMEDIA_EVENT_END)
	{
		comn::AutoCritSec lock(m_csSource);
		m_endOfStream = true;
	}
	else if (event == UMEDIA_EVENT_OPEN_FAILED)
	{
		comn::AutoCritSec lock(m_csSource);
		m_endOfStream = true;
	}

	fireMediaEvent(event);
}


void UMediaReaderSource::doCloseFile()
{
	if (m_reader)
	{
		m_reader->close();
		m_reader.reset();

		m_connected = false;
		emitMediaEvent(UMEDIA_EVENT_CLOSED);
	}

	m_isLive = false;
	m_seekable = false;
    m_cutoutTime = 0;
}

void UMediaReaderSource::dispatchPacket(UMediaPacket& pkt)
{
    if (UMediaReaderSource::DEBUG_ENABLED)
    {
        if (pkt.isVideo())
        {
            if (pkt.isKey())
            {
                CLog::debug("dispatchPacket(%p). video . size: %d, pts: %lld\n", this, pkt.size(), pkt.pts());
            }
        }
        //else
        //{
        //    CLog::debug("dispatchPacket(%p). audio . size: %d, pts: %lld, duration: %d\n", this, pkt->size, pkt->pts, pkt->duration);
        //}
    }

	fireMediaPacket(pkt);
}


int UMediaReaderSource::computeDelay(UMediaPacket& pkt)
{
	int64_t delay = 0;

	if (pkt.type() != m_refStreamIndex)
	{
		return 0;
	}

	int64_t curClock = getClock();
	util::TimePoint curPoint(curClock, pkt.pts());

	if (!m_startPoint.isSet())
	{
		m_startPoint = curPoint;
		m_lastPoint = curPoint;

		//CLog::info("*** start point. clock: %lld, pts: %lld\n", curPoint.m_clock/1000, curPoint.m_pts/1000);
	}
	else
	{
		int64_t offset = curPoint.getOffset(m_startPoint);

		int64_t ptsDelta = curPoint.m_pts - m_lastPoint.m_pts;
		int64_t clockDelta = curPoint.m_clock - m_lastPoint.m_clock;

		if (ptsDelta > MAX_SYNC_TIME)
		{
			CLog::info("----resetTimePoint %lld\n", ptsDelta / 1000);
			resetTimePoint();
		}
		else if (ptsDelta < 0)
		{
			resetTimePoint();
		}
		else
		{
			delay = offset;
		}

		m_lastPoint = curPoint;
	}

	//const char* name = (pkt->type == 0) ? "video" : "audio";
	//CLog::debug("computeDelay. packet(%s) clock:%lld, pts:%lld, delay:%d\n",
	//	name, curPoint.m_clock/1000, curPoint.m_pts/1000, delay/1000);

	return (int)delay / 1000;
}

int64_t UMediaReaderSource::getClock()
{
	int64_t clk = util::TimeHelper::getClock();
	if ((m_scale.num > 0) && (m_scale.num != m_scale.den))
	{
		clk = av_rescale(clk, m_scale.num, m_scale.den);
	}
	return clk;
}


int UMediaReaderSource::step()
{
	postCommand(kStep);
	return 0;
}

int UMediaReaderSource::nextKeyFrame()
{
	postCommand(kNextKeyFrame);
	return 0;
}


void UMediaReaderSource::delay(int ms)
{
	if (ms < MIN_DELAY_TIME)
	{
		return;
	}

	timedwait(ms);
}

void UMediaReaderSource::resetTimePoint()
{
	comn::AutoCritSec lock(m_csSource);

	m_startPoint.reset();
	m_lastPoint.reset();
}


bool UMediaReaderSource::readUntilVideoFrame()
{
	bool found = false;
	while (!found)
	{
		UMediaPacket pkt;

		int rc = read(pkt);
		if (rc == 0)
		{
			if (pkt.isVideo())
			{
				found = true;
			}

			dispatchPacket(pkt);
		}
		else if (rc == AVERROR_EOF)
		{
			break;
		}
	}

	return	found;
}

bool UMediaReaderSource::readUntilVideoKeyFrame()
{
	bool found = false;
	while (!found)
	{
		UMediaPacket pkt;

		int rc = read(pkt);
		if (rc == 0)
		{
			if (pkt.isVideo())
			{
				if (pkt.isKey())
				{
					dispatchPacket(pkt);
					found = true;
				}
			}
		}
		else if (rc == AVERROR_EOF)
		{
			break;
		}
	}

	return	found;
}

int UMediaReaderSource::read(UMediaPacket& packet)
{
	int rc = 0;
	rc = m_reader->read(packet);

	if (rc == 0)
	{
		m_curTime = packet.pts();
	}

	return rc;
}

void UMediaReaderSource::reopen()
{
	doCloseFile();

	handleOpen();
}

UMediaReaderPtr UMediaReaderSource::createReader(const std::string& url, bool loopEnabled)
{
    UMediaReaderPtr reader;

	FfmpegMediaReader* mediaReader = new FfmpegMediaReader();
	mediaReader->customAudioTs(m_audioTsCustomed);

	if (loopEnabled)
	{
		reader.reset(new LoopMediaReader(mediaReader, true, m_loopCount));
	}
	else
	{
		reader.reset(mediaReader);
	}

    return reader;
}

void UMediaReaderSource::setRetryTime(int ms)
{
    m_retryDelay = ms;
}

void UMediaReaderSource::customAudioTs(bool enabled)
{
    m_audioTsCustomed = enabled;
}

bool UMediaReaderSource::delayWait(int ms)
{
	return m_eventDelay.timedwait(ms);
}